package com.yj.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * 消费者测试类
 */
@Slf4j
@Component
public class ConsumerTest {
    /**
     * 消费者
     * DefaultMQPushConsumer和DefaultMQPullConsumer的区别是：
     * push由RocketMQ服务器主动下发消息给消费者，而pull是由消费者自己去拉取RocketMQ的消息
     * 实际上push也是由消费者定时pull，只是这个动作由client定时执行，在使用者看来就和push一样的效果
     */
    private DefaultMQPushConsumer consumer;
    /**
     * 消费组名称，同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。必须订阅完全相同的Topic。
     * RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。
     * 在集群消费下，同一条消息只会被相同消费者分组的一个消费者所消费。
     * 在广播消费下，同一条消息会被相同消费者分组的所有消费者所消费。
     * 在当前示例里，我们采用的是DefaultMQPushConsumer的默认消费方式，集群消费。
     */
    private String consumerGroup = "consumer-test";
    /**
     * RocketMQ Namesrv的地址
     */
    private String mqAddr = "127.0.0.1:9876";
    /**
     * 订阅的topic，多个topic之间用","分隔
     */
    private String topic = "test_topic_0,test_topic_1,test_topic_2";
    /**
     * 订阅的tag，不同topic之间的tag用","分隔，相同topic下的tag以"||"分隔
     */
    private String tag = "test_tag_0||test_tag_1,test_tag_3,*";

    /**
     * 根据topic和tag订阅指定消息并启动监听
     */
    @PostConstruct
    private void listenMessage() {
        log.info("consumer starting");
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(mqAddr);
        // 设置初始消费进度，如果已有消费进度则继续使用
        // CONSUME_FROM_FIRST_OFFSET ：每个 Topic 队列的第一条消息。
        // CONSUME_FROM_LAST_OFFSET ：每个 Topic 队列的最后一条消息。
        // CONSUME_FROM_TIMESTAMP ：每个 Topic 队列的指定时间开始的消息。
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        splitAndSubscribe();
        // 注册监听器：MessageListenerConcurrently多线程并发监听器、MessageListenerOrderly单线程顺序监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                if (list == null || list.isEmpty()) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                log.info("MessageExtList:{}", list);
                MessageExt ext = list.get(0);
                if (ext == null || ext.getBody() == null || ext.getBody().length == 0) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                log.info("consumer消费消息:{}", new String(ext.getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        try {
            consumer.start();
            log.info("consumer start success.");
        } catch (MQClientException e) {
            log.error("consumer start failed. msg:{}", e.getErrorMessage());
        }
    }

    /**
     * 拆分topic并订阅
     */
    public void splitAndSubscribe() {
        try {
            String[] topics = topic.split(",");
            String[] tags = tag.split(",");
            if (topics.length != tags.length) {
                String errorMsg = String.format("The number of topic and tag does not match. topic:%s, tag:%s", topic, tag);
                log.warn(errorMsg);
                throw new IllegalArgumentException(errorMsg);
            }
            for (int i = 0; i < topics.length; i++) {
                Assert.hasText(topics[i], "topic not blank");
                Assert.hasText(tags[i], "tag not blank");
            }
            for (int i = 0; i < topics.length; i++) {
                consumer.subscribe(topics[i], tags[i]);
                log.info("subscribe -> topic:{}, tag:{}", topics[i], tags[i]);
            }
        } catch (MQClientException e) {
            log.error("subscribe failed. topic:{}, tag:{}, error message:{}", topic, tag, e.getErrorMessage());
        }
    }
}